package com.example.rocketmq.demo.consumer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 延时消息消费者
 */
@Slf4j
@Component
public class DelayConsumer {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    private DefaultMQPushConsumer consumer;
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(RocketMQConstant.CONSUMER_GROUP_DELAY);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(RocketMQConstant.TOPIC_DELAY, RocketMQConstant.TAG_DELAY);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                long deliveryTimestamp = System.currentTimeMillis();
                long storeTimestamp = msg.getStoreTimestamp();
                long delayTime = deliveryTimestamp - storeTimestamp;

                log.info("接收到延时消息: msgId={}, topic={}, tags={}, content={}, 存储时间={}, 投递时间={}, 延迟={}ms",
                        msg.getMsgId(), msg.getTopic(), msg.getTags(), content,
                        sdf.format(new Date(storeTimestamp)),
                        sdf.format(new Date(deliveryTimestamp)),
                        delayTime);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        log.info("延时消息消费者启动成功");
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("延时消息消费者已关闭");
        }
    }
}
